package com.infore.logs.main.kafka;

import cn.hutool.json.JSONUtil;
import com.infore.logs.main.constant.GatewayConstant;
import com.infore.logs.main.service.TOthGatewayRequestLogService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import com.infore.common.model.entity.logs.TOthGatewayRequestLog;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Properties;

/**
 * <p> 消费者 - 接收消息 </p>
 *
 * @author lijh
 * @description 监听kafka消息
 * @date 2023/11/14 11:34
 */
@Slf4j
@Component
public class Consumer {
    @Value("${spring.kafka.bootstrap-servers}")
    private String servers;
    @Value("${spring.kafka.consumer.group-id}")
    private String groupId;

    @Autowired
    private TOthGatewayRequestLogService gatewayRequestLogService;

    @KafkaListener(topics = GatewayConstant.KAFKA_TOPIC_LOG, containerFactory = "batchConsumeFactory")
    public void consumeMessages(List<ConsumerRecord<String, String>> datas) {
        List<TOthGatewayRequestLog> list = new ArrayList<>();
        for (ConsumerRecord<String, String> record: datas) {
            try {
                TOthGatewayRequestLog requestLog = JSONUtil.toBean(record.value(), TOthGatewayRequestLog.class);
                requestLog.setCreatedTime(new Date());
                list.add(requestLog);
            } catch (Exception e) {
                // 调整偏移量到下一个位置
                Properties props = new Properties();
                props.setProperty("bootstrap.servers", servers);
                props.setProperty("group.id", groupId);
                KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
                TopicPartition topicPartition = new TopicPartition(record.topic(), record.partition());
                consumer.seek(topicPartition, record.offset() + 1);
                throw new RuntimeException(e);
            }
        }
        log.info("批量插入数据=="+list.size());
        log.info(JSONUtil.toJsonStr(list));
        gatewayRequestLogService.insertBatch(list);
    }

/*    @KafkaListener(topics = GatewayConstant.KAFKA_TOPIC_LOG)
    public void listen(ConsumerRecord<?, ?> record) {
        try {
            log.info("topic: " + record.topic() + "  <|============|>  消息内容：" + record.value());
        } catch (Exception e) {
            // 调整偏移量到下一个位置
            // 如果需要跳过当前消息，可以将偏移量 + 1
            // 如果需要跳过一段消息，可以将偏移量设置为一个特定值
            // 例如，将偏移量设置为 record.offset() + n，其中 n 是要跳过的消息数量
            // 调整消费者的偏移量到下一个可用位置
            // 设置消费者的配置参数
            Properties props = new Properties();
            props.setProperty("bootstrap.servers", servers);
            props.setProperty("group.id", groupId);
            KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
            TopicPartition topicPartition = new TopicPartition(record.topic(), record.partition());
            consumer.seek(topicPartition, record.offset() + 1);
            throw new RuntimeException(e);
        }
    }*/
}
